/*
 *  Filename:lsv_log_gc_old_storage.c
 *  Description:
 *
 *  Created on: 2017年3月27日
 *  Author: Asdf(825674301)
 */
#include "lsv_conf.h"
#include "lsv_help.h"
#include "lsv_log.h"
#include "lsv_volume.h"
#include "lsv_gc_internal.h"

int lsv_gc_old_storage_meta_write(lsv_volume_proto_t *lsv_info,
                                      lsv_gc_old_storage_meta_t *meta,
                                      lsv_gc_old_storage_bitmap_meta_t *bitmap_meta) {
        (void) lsv_info;
        (void) meta;
        (void) bitmap_meta;
        return 0;
}

int lsv_gc_old_storage_meta_read(lsv_volume_proto_t *lsv_info,
                                     lsv_gc_old_storage_meta_t *meta,
                                     lsv_gc_old_storage_bitmap_meta_t *bitmap_meta) {
        (void) lsv_info;
        (void) meta;
        (void) bitmap_meta;
        return 0;
}

static int __lsv_gc_old_storage_add(lsv_volume_proto_t *lsv_info, lsv_u32_t chunk_id, int force) {
        int ret;
        lsv_log_info_t *log_info = NULL;
        lsv_gc_context_t *gc_context = NULL;
        lsv_gc_old_storage_t *old_storage = NULL;
        lsv_u32_t old_storage_chkid;

        log_info = lsv_info->log_info;
        gc_context = log_info->gc_context;
        old_storage = &gc_context->old_storage;

        DINFO("storage a log data,chunk_id:%u\n", chunk_id);

        lsv_lock(&old_storage->lock);

        if (chunk_id != LSV_CHUNK_NULL) {
                old_storage->ready_chunk.storage_chkid[old_storage->ready_chunk.count++] = chunk_id;
        }

        if (old_storage->ready_chunk.count == LSV_GC_OS_CHUNK_NUM) {
                // 先分配chunk_id, 后才会使用，方便建立单链表的next链
                DINFO("old storage a chunk data, chunk_id %u\n", old_storage->ready_chkid);
                ret = lsv_volume_chunk_malloc(lsv_info, LSV_GC_STORAGE_TYPE, &old_storage_chkid);
                if (unlikely(ret)) {
                        YASSERT(0);
                }

                lsv_volume_io_t vio;
                lsv_volume_io_init(&vio, old_storage->ready_chkid, 0, LSV_CHUNK_SIZE, LSV_GC_STORAGE_TYPE);
                old_storage->ready_chunk.next_chkid = old_storage_chkid;
                ret = lsv_volume_chunk_update(lsv_info, &vio, (lsv_s8_t *) &old_storage->ready_chunk);
                if (ret < 0) {
                        YASSERT(0);
                }

                old_storage->chunk_count++;

                // clean stack
                old_storage->ready_chkid = old_storage_chkid;
                old_storage->ready_chunk.count = 0;
                old_storage->ready_chunk.next_chkid = LSV_CHUNK_NULL;
        }

        if (force) {
#if 0
                DINFO("flush ready chunk, count %u\n", old_storage->ready_chunk.count);
                YASSERT(old_storage->ready_chunk.count < LSV_GC_OS_CHUNK_NUM);

                lsv_volume_io_t vio;
                lsv_volume_io_init(&vio, old_storage->ready_chkid, 0, LSV_CHUNK_SIZE, LSV_GC_STORAGE_TYPE);
                old_storage->ready_chunk.next_chkid = LSV_CHUNK_NULL;
                ret = lsv_volume_chunk_update(lsv_info, &vio, (lsv_s8_t *) &old_storage->ready_chunk);
                if (ret < 0) {
                        YASSERT(0);
                }
#endif

                // 包含了flush ready chunk的过程
                lsv_gc_old_storage_head_set(lsv_info);
        }

        lsv_unlock(&old_storage->lock);

        lsv_gc_stat(lsv_info);
        return 0;
}

/**
 * @note 先批量添加到缓冲区，然后提交到os，整个过程要原子化
 *
 * @param lsv_info
 * @param chunk_id
 * @param force if true, flush ready chunk
 * @return
 */
int lsv_gc_old_storage_remove_and_add(lsv_volume_proto_t *lsv_info, lsv_u32_t chunk_id, int force) {
        int ret = 0;
        lsv_log_info_t *log_info = NULL;
        lsv_gc_context_t *gc_context = NULL;
        lsv_gc_logctrl_t *logctrl;

        if (chunk_id != LSV_CHUNK_NULL) {
                log_info = lsv_info->log_info;
                gc_context = log_info->gc_context;

                ret = hash_table_remove(gc_context->logctrl_tab, &chunk_id, (void **) &logctrl);

                YASSERT(0 == ret);
                YASSERT(logctrl != NULL);

                // TODO why unset bitmap?
                lsv_gc_logctrl_free(lsv_info, logctrl);
        }

        __lsv_gc_old_storage_add(lsv_info, chunk_id, force);

        return 0;
}

/**
 * @todo 两次IO
 *
 * @param lsv_info
 * @return
 */
int lsv_gc_old_storage_head_set(lsv_volume_proto_t *lsv_info) {
        int ret = 0;
        lsv_log_info_t *log_info = NULL;
        lsv_gc_context_t *gc_context = NULL;
        lsv_gc_old_storage_t *old_storage = NULL;
        lsv_volume_io_t vio;

        log_info = lsv_info->log_info;
        gc_context = log_info->gc_context;
        old_storage = &gc_context->old_storage;

        YASSERT(old_storage->ready_chunk.count < LSV_GC_OS_CHUNK_NUM);

        lsv_volume_io_init(&vio,
                           old_storage->ready_chkid,
                           0,
                           (2 + old_storage->ready_chunk.count) * sizeof(lsv_u32_t),
                           LSV_GC_STORAGE_TYPE);
        ret = lsv_volume_chunk_update(lsv_info, &vio, (lsv_s8_t *)&old_storage->ready_chunk);
        if (ret < 0) {
                DERROR("store old_storage err, ret %d\n", ret);
                YASSERT(0);
        }

        lsv_gc_old_storage_meta_t meta;

        meta.chunk_count = old_storage->chunk_count;
        meta.gc_chkid = old_storage->gc_chkid;
        meta.ready_chkid = old_storage->ready_chkid;
        meta.ready_chunk_count = old_storage->ready_chunk.count;

        lsv_volume_io_init(&vio,
                           LSV_PRIM_CHUNK_ID,
                           lsv_info->u.gc_os_page_id * LSV_PAGE_SIZE + LSV_GC_OS_HEAD,
                           sizeof(meta),
                           LSV_GC_STORAGE_TYPE);
        ret = lsv_volume_chunk_update(lsv_info, &vio, (lsv_s8_t *)&meta);
        if (ret < 0) {
                DERROR("set old stroage err, ret %d\n", ret);
                YASSERT(0);
        }

        DINFO("set old stroage, chunk_count:%u,ready_count:%u\n",
              gc_context->old_storage.chunk_count,
              gc_context->old_storage.ready_chunk.count);
        return 0;
}
